package com.imchenio.utils.parallel;

import com.imchenio.utils.parallel.runnable.ChannelRunnable;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ProcessChannel<T>{

    private static AtomicInteger count = new AtomicInteger(0);

    private int ringNode;

    private final BlockingQueue<T> blockingQueue;

    private final ScheduledExecutorService service;

    public ProcessChannel(int ringNode, int blockingQueueSize, ChannelRunnable<T> runnable) {
        ringNode = ringNode;

        blockingQueue = new LinkedBlockingQueue<>(blockingQueueSize);
        runnable.setBlockingQueue(blockingQueue);

        service = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("process-channel-" + count.incrementAndGet());
                return thread;
            }
        }, new ThreadPoolExecutor.CallerRunsPolicy());
        // 是否会导致BlockingQueue堆积需要确认下
        service.scheduleAtFixedRate(runnable, 0, 200, TimeUnit.MILLISECONDS);
    }

    public boolean offer(T object) {
        return blockingQueue.offer(object);
    }

    public boolean offer(T object, long x, TimeUnit unit) throws InterruptedException {
        return blockingQueue.offer(object, x, unit);
    }
}
